package com.atguigu.gmall.realtime.app.dws;

import com.atguigu.gmall.realtime.app.udf.KeywordUDTF;
import com.atguigu.gmall.realtime.bean.KeywordStats;
import com.atguigu.gmall.realtime.common.GmallConstant;
import com.atguigu.gmall.realtime.utils.ClickHouseUtil;
import com.atguigu.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class KeywordStatsApp {
    public static void main(String[] args) throws Exception {
        //TODO 0 基本环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(3);

        //CK设置

        //TODO 1 定义Table流环境
        EnvironmentSettings settings = EnvironmentSettings.newInstance()
                .inStreamingMode()
                .build();

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env,settings);


        // TODO 2 注册自定义函数
        tableEnv.createTemporarySystemFunction("ik_analyse", KeywordUDTF.class);

        // TODO 3 数据源定义
        String groupId = "keyword_stats_app";
        String pageViewSourceTopic = "dwd_page_log";

        // 水印延迟时间2秒
        tableEnv.executeSql("CREATE TABLE page_view " +
                "(common MAP<STRING,STRING>, " +
                "page MAP<STRING,STRING>, ts BIGINT, " +
                "rowtime AS TO_TIMESTAMP(FROM_UNIXTIME(ts/1000, 'yyyy-MM-dd HH:mm:ss')) , "+
                "WATERMARK FOR rowtime AS rowtime - INTERVAL '2' SECOND) " +
                "WITH ( "+ MyKafkaUtil.getKafkaDDL(pageViewSourceTopic,groupId) + ")");

        //TODO 4 过滤数据
        Table fullwordView = tableEnv.sqlQuery("select page['item'] fullword," +
                "rowtime from page_view " +
                "where page['page_id']='good_list' " +
                "and page['item'] IS NOT NULL");

        //TODO 5 利用UDTF进行拆分
        Table keywordView = tableEnv.sqlQuery("select keyword,rowtime from " + fullwordView + "," +
                "LATERAL TABLE(ik_analyse(fullword)) as T(keyword)");

        //TODO 6 聚合
        Table keywordStatsSearch = tableEnv.sqlQuery("select keyword,count(*) ct , '" +
                GmallConstant.KEYWORD_SEARCH + "' source, " +
                "DATE_FORMAT(TUMBLE_START(rowtime, INTERVAL '10' SECOND), 'yyyy-MM-dd HH:mm:ss') stt," +
                "DATE_FORMAT(TUMBLE_END(rowtime, INTERVAL '10' SECOND), 'yyyy-MM-dd HH:mm:ss') edt, " +
                "UNIX_TIMESTAMP()*1000 ts from " + keywordView +
                "  GROUP BY TUMBLE(rowtime,INTERVAL '10' SECOND),keyword ");

        DataStream<KeywordStats> keywordStatsDataStream = tableEnv.<KeywordStats>toAppendStream(keywordStatsSearch, KeywordStats.class);


        keywordStatsDataStream.print("search >>>> ");
        keywordStatsDataStream.addSink(ClickHouseUtil.<KeywordStats>getJdbcSink(
                "insert into keyword_stats_2021 (keyword,ct,source,stt,edt,ts) values (?,?,?,?,?,?)"
        ));


        env.execute();
    }
}
